package com.client;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.nio.channels.spi.SelectorProvider;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;

import com.common.utilities.ChangeRequest;
import com.common.utilities.ChangeRequestType;

/**
 * User: wattsc5
 * <p/>
 * Date: 2/28/12
 */
public class NioClient implements Runnable
{
	private static final Logger logger = Logger.getLogger(NioClient.class);

	private int port;

	private InetAddress hostAddress;

	private final LinkedList<ChangeRequest> pendingChanges = new LinkedList<ChangeRequest>();

	private final Map<SocketChannel, ResponseHandler> rspHandlers = Collections.synchronizedMap(new HashMap<SocketChannel, ResponseHandler>());

	private final Map<SocketChannel, List<ByteBuffer>> pendingData = new HashMap<SocketChannel, List<ByteBuffer>>();

	private Selector selector;

	private SocketChannel socket;

	private static final int MaxBufferSize = 1024;

	private final ByteBuffer readBuffer = ByteBuffer.allocate(MaxBufferSize);

	public NioClient(InetAddress hostAddress, int port)
	{
		this.port = port;
		this.hostAddress = hostAddress;
		this.selector = this.initializeSelector();
	}

	public void run()
	{
		while (true)
		{
			try
			{
				// Process any pending changes
				synchronized (this.pendingChanges)
				{
					Iterator changes = this.pendingChanges.iterator();
					while (changes.hasNext())
					{
						ChangeRequest change = (ChangeRequest) changes.next();
						switch (change.getRequestType())
						{
							case ChangeOps:
								SelectionKey key = change.getSocket().keyFor(this.selector);
								key.interestOps(change.getOps());
								break;
							case Register:
								change.getSocket().register(this.selector, change.getOps());
								break;
						}
					}
					this.pendingChanges.clear();
				}

				// Wait for an event one of the registered channels
				this.selector.select();

				// Iterate over the set of keys for which events are available
				Iterator selectedKeys = this.selector.selectedKeys().iterator();
				while (selectedKeys.hasNext())
				{
					SelectionKey key = (SelectionKey) selectedKeys.next();
					selectedKeys.remove();

					if (!key.isValid())
					{
						continue;
					}

					// Check what event is available and deal with it
					if (key.isConnectable())
					{
						this.finishConnection(key);
					}
					else if (key.isReadable())
					{
						this.read(key);
					}
					else if (key.isWritable())
					{
						this.write(key);
					}
				}
			}
			catch (Exception e)
			{
				logger.error(e);
			}
		}
	}

	public void send(byte[] data, ResponseHandler handler)
		throws IOException
	{
		logger.trace("About to send data to socket - " + new String(data));

		if (socket == null)
		{
			socket = this.initializeConnection();
			this.rspHandlers.put(socket, handler);
		}
		else
		{
			synchronized (this.pendingChanges)
			{
				this.pendingChanges.add(new ChangeRequest(socket, ChangeRequestType.ChangeOps, SelectionKey.OP_WRITE));
			}
		}

		synchronized (this.pendingData)
		{
			List<ByteBuffer> queue = this.pendingData.get(socket);

			if (queue == null)
			{
				queue = new ArrayList<ByteBuffer>();
				this.pendingData.put(socket, queue);
			}

			queue.add(ByteBuffer.wrap(data));
		}

		this.selector.wakeup();
	}

	private void handleResponse(SocketChannel socketChannel, byte[] data)
		throws IOException
	{
		// Look up the handler for this channel
		ResponseHandler handler = this.rspHandlers.get(socketChannel);

		// And pass the response to it
		if (handler.handleResponse(socketChannel, data))
		{
			// The handler has seen enough, close the connection
			socketChannel.close();
			socketChannel.keyFor(this.selector).cancel();
		}
	}

	private void read(SelectionKey key)
		throws IOException
	{
		SocketChannel socketChannel = (SocketChannel) key.channel();

		// Clear out our read buffer so it's ready for new data
		this.readBuffer.clear();

		// Attempt to read off the channel
		int numberOfBytesRead;
		try
		{
			numberOfBytesRead = socketChannel.read(this.readBuffer);
		}
		catch (IOException e)
		{
			// The remote forcibly closed the connection, cancel
			// the selection key and close the channel.
			key.cancel();
			socketChannel.close();
			logger.debug("Server disconnected forcibly");
			return;
		}

		if (numberOfBytesRead == -1)
		{
			// Remote entity shut the socket down cleanly. Do the
			// same from our end and cancel the channel.
			key.channel().close();
			key.cancel();
			logger.debug("Server disconnected cleanly");
			return;
		}

		// Make a copy of the data here so we can output it to the logger then send it to the game worker.
		byte[] copyData = new byte[numberOfBytesRead];
		System.arraycopy(this.readBuffer.array(), 0, copyData, 0, numberOfBytesRead);

		logger.trace("Data received - " + socketChannel.socket().getInetAddress() + ":" + socketChannel.socket().getPort() + " - " + new String(copyData));

		// Handle the response
		this.handleResponse(socketChannel, copyData);
	}

	private void write(SelectionKey key)
		throws IOException
	{
		SocketChannel socketChannel = (SocketChannel) key.channel();

		synchronized (this.pendingData)
		{
			List queue = (List) this.pendingData.get(socketChannel);

			// Write until there's not more data ...
			while (!queue.isEmpty())
			{
				ByteBuffer buf = (ByteBuffer) queue.get(0);
				logger.trace("Writing to socket - " + socketChannel.socket().getInetAddress() + ":" + socketChannel.socket().getPort());

				socketChannel.write(buf);
				if (buf.remaining() > 0)
				{
					// ... or the socket's buffer fills up
					break;
				}
				queue.remove(0);
			}

			if (queue.isEmpty())
			{
				// We wrote away all data, so we're no longer interested
				// in writing on this socket. Switch back to waiting for
				// data.
				key.interestOps(SelectionKey.OP_READ);
			}
		}
	}

	private void finishConnection(SelectionKey key)
		throws IOException
	{
		SocketChannel socketChannel = (SocketChannel) key.channel();

		// Finish the connection. If the connection operation failed
		// this will raise an IOException.
		try
		{
			socketChannel.finishConnect();
			logger.debug("Client connected to - " + this.hostAddress + ":" + this.port);
		}
		catch (IOException e)
		{
			// Cancel the channel's registration with our selector
			logger.warn("Client could not connect to " + this.hostAddress + ":" + this.port);
			key.cancel();
			return;
		}

		// Register an interest in writing on this channel
		key.interestOps(SelectionKey.OP_WRITE);
	}

	private SocketChannel initializeConnection()
		throws IOException
	{
		// Create non blocking channel
		SocketChannel socketChannel = SocketChannel.open();
		socketChannel.configureBlocking(false);

		// Start connection establishment
		socketChannel.connect(new InetSocketAddress(this.hostAddress, this.port));

		// Queue a channel registration since the caller is not the selected thread
		// As part of the registration we'll register an interest in connection events.
		// These are raised when a channel is ready to complete connection establishment.
		synchronized (this.pendingChanges)
		{
			this.pendingChanges.add(new ChangeRequest(socketChannel, ChangeRequestType.Register, SelectionKey.OP_CONNECT));
		}

		return socketChannel;
	}

	/**
	 * Creates a new non-blocking selector.
	 *
	 * @return the new selector.
	 */
	private Selector initializeSelector()
	{
		logger.trace("Initializing client selector");
		Selector selector = null;
		try
		{
			// Create a new selector
			selector = SelectorProvider.provider().openSelector();
		}
		catch (IOException e)
		{
			logger.error(e);
		}

		return selector;
	}
}
